// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/lbank/feed_v2/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/lbank/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::lbank::feed_v2::impl {

void LbankFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());

  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsObject()) {
    std::string str_type;

    if (document.HasMember("pong")) {
      return;
    }

    if (document.HasMember("type") && document["type"].IsString()) {
      str_type = document["type"].GetString();

      if (document.HasMember("count")) {
        str_type += document["count"].GetString();
      }

      if (document.HasMember("kbar") && document["kbar"].HasMember("slot")) {
        str_type += document["kbar"]["slot"].GetString();
      }
    } else if (document.HasMember("url")) {
      const std::string& url = document["url"].GetString();
      if (url.find("/v2/kline.do") != std::string::npos) {
        str_type = "kline_rest";
      }
    }

    const auto& channel = LbankChannel::Parse(str_type);
    if (!CheckChannelFilter(channel.channel)) {
      return;
    }

    std::string native_symbol;
    int64_t ts;
    if (document.HasMember("pair")) {
      native_symbol = document["pair"].GetString();
      ts = TimestampFromIso8601(std::string(document["TS"].GetString()) + "Z");
      ts -= 28800000000000LL;  // Beijing Time -> UTC
    }

    upd_mgr_.onBeginPacket(packet);

    if (channel.channel == "books" || channel.channel == "books50") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        LbankFeedUpdateParser::ParseBook(document["depth"], book, ts, BookBuilder::SNAPSHOT);
        upd_mgr_.PublishBook();
      }
    } else if (channel.channel == "increment_depth") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        LbankFeedUpdateParser::ParseBook(document["incrdepth"], book, ts, BookBuilder::DELTA);
        upd_mgr_.PublishBook();
      }
    } else if (channel.channel == "trade") {
      auto* trade = upd_mgr_.GetTradeMutable();
      if (trade != nullptr) {
        LbankFeedUpdateParser::ParseTrade(document["trade"], trade);
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (
        channel.channel == "kline_ws_1min" || channel.channel == "kline_ws_5min" ||
        channel.channel == "kline_ws_15min" || channel.channel == "kline_ws_1hour" ||
        channel.channel == "kline_ws_1day") {
      // kline
      auto interval = LbankFeedUpdateParser::ParseKlineInterval(channel.channel);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        klines->UpdateClear();
        if (LbankFeedUpdateParser::Parse1Kline(document["kbar"], klines, ts, interval)) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    } else if (channel.channel == "kline_rest") {
      auto parameters =
          base::feed::FeedParsingProcessor::GetRequestParameters(document["url"].GetString());

      native_symbol = parameters["symbol"];
      const std::string& kline_interval = parameters["type"];
      auto interval = LbankFeedUpdateParser::ParseKlineInterval(kline_interval);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        const auto& data = document["data"]["data"].GetArray();
        std::optional<int64_t> first_open_timestamp;
        first_open_timestamp = stoll(data[0][0].GetString()) * 1'000'000'000LL;

        if (klines->Size() == 0) {
          klines->SnapshotClear();
        } else if (
            first_open_timestamp && first_open_timestamp < klines->RBegin()->open_timestamp) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        bool to_publish = false;
        for (const auto& kline_data : data) {
          bool freezed =
              LbankFeedUpdateParser::ParseKlineRest(kline_data, klines, interval, packet.timestamp);
          to_publish = to_publish || freezed;
        }
        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    } else {
      LOG(INFO) << "[Lbank Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload;
    }

    upd_mgr_.onEndPacket(packet);
  }
}
}  // namespace coin2::exchange::lbank::feed_v2::impl
